package com.thiswhyme.mq.kafka.consumer1.consumer;

import com.thiswhyme.mq.kafka.common.consts.TopicNameDefinition;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

/**
 * @description: 定时消费者
 * @author: thiswhyme
 * @date: 2021/6/3
 */
@Slf4j
@Component
@AllArgsConstructor
@EnableScheduling
public class SchedulerConsumer {

    private static final String CONSUMER_ID_TIMING = "timingConsumer";

    /**
     * "@KafkaListener"注解所标注的方法并不会在IOC容器中被注册为Bean，而是会被注册在KafkaListenerEndpointRegistry中
     */
    private final KafkaListenerEndpointRegistry registry;

    /**
     * 定时监听
     * @param record
     */
    @KafkaListener(id = CONSUMER_ID_TIMING, idIsGroup = false, topics = TopicNameDefinition.TIMING_TOPIC, containerFactory = "delayContainerFactory")
    public void onTimingMessage(ConsumerRecord<?, ?> record){
        log.info("=====【onTimerMessage】监听到消息，topic：{}，partition：{}，内容：{}", record.topic(),
                record.partition(), record.value());
    }

    /**
     * 定时启动监听器
     */
    @Scheduled(cron = "0 * * * * ?")
    public void startListener() {
        if (!registry.getListenerContainer(CONSUMER_ID_TIMING).isRunning()) {
            log.info("=====启动监听器...");
            registry.getListenerContainer(CONSUMER_ID_TIMING).start();
        }
    }

    /**
     * 定时暂停监听器
     */
    @Scheduled(cron = "10,30 * * * * ?")
    public void pauseListener() {
        if (!registry.getListenerContainer(CONSUMER_ID_TIMING).isContainerPaused()) {
            log.info("=====暂停监听器...");
            registry.getListenerContainer(CONSUMER_ID_TIMING).pause();
        }
    }

    /**
     * 定时恢复监听器
     */
    @Scheduled(cron = "20,40 * * * * ?")
    public void resumeListener() {
        if (registry.getListenerContainer(CONSUMER_ID_TIMING).isContainerPaused()) {
            log.info("=====恢复监听器...");
            registry.getListenerContainer(CONSUMER_ID_TIMING).resume();
        }
    }

    /**
     * 定时停止监听器
     */
    @Scheduled(cron = "50 * * * * ?")
    public void stopListener() {
        if (registry.getListenerContainer(CONSUMER_ID_TIMING).isRunning()) {
            log.info("=====关闭监听器...");
            registry.getListenerContainer(CONSUMER_ID_TIMING).stop();
        }
    }
}
